package com.example.hadoopdemo.executor.weather;

import com.example.hadoopdemo.config.HadoopConfig;
import com.example.hadoopdemo.props.HadoopProperties;
import lombok.AllArgsConstructor;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.springframework.stereotype.Component;

/**
 * Map/Reduce工具类
 *
 * @author linhaiy
 * @date 2019.05.18
 */
@Component
@AllArgsConstructor
public class WeatherJobsUtils {
    private HadoopProperties hadoopProperties;
//    @Value("${hdfs.path}")
//    private String path;
//    private static String hdfsPath;
    /**
     * 获取HDFS配置信息
     * @return
     */
//    public static Configuration getConfiguration() {
//        Configuration configuration = new Configuration();
//        configuration.set("fs.defaultFS", hdfsPath);
//        configuration.set("mapred.job.tracker", hdfsPath);
//        // 运行在yarn的集群模式
//        // configuration.set("mapreduce.framework.name", "yarn");
//        // 这个配置是让main方法寻找该机器的mr环境
//        // configuration.set("yarn.resourcemanmager.hostname", "node1");
//        return configuration;
//    }

    /**
     * 获取单词一年最高气温计算配置
     * @param jobName
     * @return
     */
    public JobConf getWeatherJobsConf(String jobName) {
//        JobConf jobConf = new JobConf(getConfiguration());
        HadoopConfig hadoopConfig = new HadoopConfig();
        JobConf jobConf = new JobConf(hadoopConfig.getConfiguration(hadoopProperties));
        jobConf.setJobName(jobName);
        jobConf.setOutputKeyClass(Text.class);
        jobConf.setOutputValueClass(LongWritable.class);
        jobConf.setMapperClass(com.example.hadoopdemo.executor.weather.WeiboMap.class);
        jobConf.setReducerClass(com.example.hadoopdemo.executor.weather.WeiboReduce.class);
        jobConf.setInputFormat(TextInputFormat.class);
        jobConf.setOutputFormat(TextOutputFormat.class);
        return jobConf;
    }
//    @PostConstruct
//    public void getPath() {
//        hdfsPath = this.path;
//    }
//    public static String getHdfsPath() {
//        return hdfsPath;
//    }
}
